package org.cloudplatform.common.feign.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.cloudplatform.common.feign.register.AbstractInstanceRegister;

import java.util.Properties;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 类描述：
 * 作者：徐卫超 (cc)
 * 时间 2022/5/13 15:29
 */

@Slf4j
public class ServiceInstanceZookeeper extends AbstractInstanceRegister implements SubscriptionClient {
    // zookeeper 连接客户端
    private volatile CuratorFramework curatorFramework = null;
    private String path = null;
    private final ScheduledThreadPoolExecutor scheduled = new ScheduledThreadPoolExecutor(1);
    private ServiceCuratorCacheListener serviceCuratorCacheListener;

    /**
     * 连接到zookeeper客户端
     *
     * @param properties address: zookeeper 连接地址( 192.168.0.128:2181,192.168.0.129:2181,192.168.0.130:2181 )
     *                   sessionTimeout: session超时时间 默认：6000
     *                   connectionTimeout: 连接超时时间 默认：3000
     */
    @Override
    public synchronized void connect(Properties properties) {
        if (this.curatorFramework != null) {
            log.warn("zookeeper 已连接 无需再次连接");
        }
        this.curatorFramework = CuratorFrameworkFactory.builder()
                .connectString(properties.getProperty("address", "192.168.101.22:2181"))
                .sessionTimeoutMs(Integer.parseInt(properties.getProperty("sessionTimeout", "6000")))
                .connectionTimeoutMs(Integer.parseInt(properties.getProperty("connectionTimeout", "3000")))
                .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
        this.curatorFramework.start();
        log.info("zookeeper 连接成功");
    }

    /**
     * 关闭订阅信息
     */
    @Override
    public synchronized void close() {
        if (curatorFramework == null) {
            log.warn("zookeeper 已断开连接，无需再次断开连接");
        }
        scheduled.shutdownNow();
        this.curatorFramework.close();
        this.curatorFramework = null;
    }


    /**
     * 订阅微服务
     *
     * @param path 微服务的根节点信息
     */
    @Override
    public synchronized void subscribe(String path) {
        if (path == null) {
            throw new RuntimeException("无效订阅路径");
        }
        if (this.path != null) {
            throw new RuntimeException("已经订阅了服务无法再次订阅");
        }
        this.path = path;

        if (this.curatorFramework == null) {
            throw new RuntimeException("还未连接到zk服务器");
        }
        assert curatorFramework.getState().equals(CuratorFrameworkState.STARTED);
        try {
            curatorFramework.blockUntilConnected();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        doSubscribe(path);
    }

    @Override
    public synchronized void completed() {
        if (serviceCuratorCacheListener == null) {
            throw new RuntimeException("还未订阅数据 请先订阅数据");
        }
        this.serviceCuratorCacheListener.isDone();
    }

    /**
     * 订阅微服务
     *
     * @param path 微服务的根节点信息
     */
    private void doSubscribe(String path) {
        CuratorCache curatorCache = CuratorCache.build(curatorFramework, path);
        this.serviceCuratorCacheListener = new ServiceCuratorCacheListener(this, path, curatorFramework);
        curatorCache.listenable().addListener(serviceCuratorCacheListener, scheduled);
        curatorCache.start();
        scheduled.scheduleWithFixedDelay(serviceCuratorCacheListener, 1, 30, TimeUnit.SECONDS);
    }
}
